<?php
declare (strict_types=1);

namespace app\command;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Wire\AMQPTable;
use think\console\Command;
use think\console\Input;
use think\console\Output;

class Retry extends Command
{
    protected function configure()
    {
        // 指令配置
        $this->setName('retry')
            ->setDescription('the retry command');
    }

    protected function execute(Input $input, Output $output)
    {
        // 交换机名
        $exchange = 'dlx_exchange';
        // 建立连接
        $connection = new AMQPStreamConnection('127.0.0.1', '5672', 'guest', 'guest', '/');
        // 获取信道
        $channel = $connection->channel();
        // 声明交换机
        $channel->exchange_declare($exchange, 'direct', false, false, false);
        $channel->queue_declare('dlx_queue', false, true, false, false);
        $channel->queue_bind('dlx_queue', $exchange, 'dlx_routing_key');

        // 创建重试队列
        $channel->exchange_declare('delay_exchange', 'direct', false, false, false);

        // 消息消费
        $channel->basic_consume('dlx_queue', '', false, false, false, false, function ($msg) use ($output) {
//            $body = iconv("UTF-8", "GB2312//IGNORE", $msg->body);
            $output->writeln(date("h:i:s") . " Received " . $msg->body . PHP_EOL);

            // 确认消息处理完成
            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);

            $msg_headers = $msg->get('application_headers')->getNativeData();

            // 重试次数超过3次，则入库告警
            if (intval($msg_headers['retry_nums']) > 3) {
//                $body = iconv("UTF-8", "GB2312//IGNORE", "重试次数超过3次，则入库告警");
                $output->writeln(date("h:i:s") . " Error " . "重试次数超过3次，则入库告警" . PHP_EOL);
            } else {
                // 建立连接
                $connection = new AMQPStreamConnection('127.0.0.1', '5672', 'guest', 'guest', '/');
                // 获取信道
                $channel = $connection->channel();
                // 重试次数加1
                $headers = new AMQPTable([
                    "retry_nums" => intval($msg_headers['retry_nums']) + 1
                ]);
                $msg->set('application_headers', $headers);

                // 放回重试队列
                $channel->basic_publish($msg, "delay_exchange", "delay_routing_key");
            }
        });

        while (count($channel->callbacks)) {
            $channel->wait();
        }

        // 关闭信道
        $channel->close();
        // 关闭连接
        $connection->close();
    }
}
